package com.youi.util;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * 1. 创建可自定义线程名的ThreadFactory(via guava)
 * 2. 优雅关闭线程池,防止第三方Runnable未捕捉异常导致线程跑飞
 *
 * @description:线程池工具类
 * @Author: winty
 * @Date: 2020/5/15 14:35
 */
public class ThreadPoolUtil {

    private static final Logger logger = LoggerFactory.getLogger(ThreadPoolUtil.class);

    /**
     * cpu核心数
     */
    private static final int CPU_CORES = Runtime.getRuntime().availableProcessors();

    /**
     * 线程工厂
     */
    private static ThreadFactory threadFactory = new ThreadFactoryBuilder()
            .setNameFormat("MeiYiJiaHealth-pool-%d")
            .setDaemon(true)
            .build();

    /**
     * 自定义线程池
     */
    public static final ExecutorService executor = new ThreadPoolExecutor(
            CPU_CORES * 20, CPU_CORES * 30, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(4096), threadFactory, new ThreadPoolExecutor.AbortPolicy());


    /**
     * 异步执行，有返回值
     *
     * @param supplier
     * @param <U>
     * @return CompletableFuture<U>
     */
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
        return CompletableFuture.supplyAsync(supplier, executor);
    }

    /**
     * 异步执行，有返回值，但是返回exceptionally CompletableFuture对象
     *
     * @param supplier
     * @param <U>
     * @return
     */
    public static <U> CompletableFuture<U> supplyAsyncExceptionally(Supplier<U> supplier) {
        return CompletableFuture.supplyAsync(supplier, executor).exceptionally(e -> {
            logger.error("supplyAsync error", e);
            return null;
        });
    }

    /**
     * 异步执行，没有返回值
     *
     * @param runnable
     */
    public static void runAsync(Runnable runnable) {
        CompletableFuture.runAsync(ThreadPoolUtil.safeRunnable(runnable), executor).exceptionally(e -> {
            logger.error("runAsync error", e);
            return null;
        });
    }

    /**
     * 异步执行，没有返回值，但是返回exceptionally CompletableFuture对象
     *
     * @param runnable
     */
    public static CompletableFuture runAsyncExceptionally(Runnable runnable) {
        return CompletableFuture.runAsync(ThreadPoolUtil.safeRunnable(runnable), executor).exceptionally(e -> {
            logger.error("runAsync error", e);
            return null;
        });
    }

    /**
     * 防止用户没有捕捉异常导致中断了线程池中的线程, 使得SchedulerService无法继续执行.
     * <p>
     * 在无法控制第三方包的Runnable实现时，调用本函数进行包裹.
     */
    public static Runnable safeRunnable(Runnable runnable) {
        return new SafeRunnable(runnable);
    }

    /**
     * 线程包装类,保证不会有Exception抛出到线程池的Runnable包裹类，
     * 防止用户没有捕捉异常导致中断了线程池中的线程, 使得SchedulerService无法执行. 在无法控制第三方包的Runnable实现时，使用本类进行包裹.
     */
    public static class SafeRunnable implements Runnable {
        private Runnable runnable;

        public SafeRunnable(Runnable runnable) {
            this.runnable = runnable;
        }

        @Override
        public void run() {
            try {
                runnable.run();
            } catch (Throwable e) {
                // catch any exception, because the scheduled thread will break if the exception thrown to outside.
                logger.error("Unexpected error occurred in task", e);
            }
        }
    }


}
